#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   For_Jenkins_2_test.py
@Time    :   2022/08/31 13:31:28
@Author  :   YangLiang 
@Version :   1.0
@Contact :   yangliang@chengtech.com
@License :   (C)Copyright 2022-2025
@Desc    :   None
'''

# here put the import lib
from pickle import FALSE, TRUE
from pytest_assume.plugin import assume
 
import pytest
import time
import requests
from sympy import I
# from sympy import assuming
from commDriver.TS_CAN import *
import allure
import sys
from collections.abc import Iterable

from sqlalchemy import false
sys.dont_write_bytecode = True

import project_CAN
import Set_up


SuppressPosMsg_extended = [     ([0x10, 0x82], [0x7F, 0x10, 0x12]),
                                ([0x11, 0x81], [0x7F, 0x11, 0x12])
                          ]

SuppressPosMsg_programming = [  ([0x10, 0x81], [0x50, 0x01, 0x00, 0x32, 0x01, 0xF4]),
                                ([0x11, 0x81], [0x7F, 0x11, 0x12])
                             ]



class Test_BYD_Calibration(object):

    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD Calibration process in extended session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_BYD_Calibration_Phy(self,action,Driver):      
        iStep = 1
        # 写VIN
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '2E F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17', '6E F1 90', 3, iStep,True, 'UDS', 'noClearFault')
        # 读取VIN
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "22 F1 90", "62 F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17", 0, iStep, False, 'UDS', 'noClearFault')
        # 清除故障并读取DTC
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "19 02 09", "59 02 09", 0, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "2E 34 01 01", "6E 34 01", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "22 34 01", "62 34 01 01", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "31 01 54 DF", "71 01 54 DF", 0, iStep, False, 'UDS', 'noClearFault')
        time.sleep(1)
        for i in range(6):
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "31 03 54 DF", "71 03 54 DF", 0, iStep, False, 'UDS', 'noClearFault')  
            time.sleep(1)          

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "31 02 54 DF", "71 02 54 DF", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "2E 34 01 00", "6E 34 01", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "22 34 01", "62 34 01 00", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "11 01", "51 01", 0, iStep, False, 'UDS', 'noClearFault')
        time.sleep(3)

        # 清除故障并读取DTC
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "19 02 09", "59 02 09", 3, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)

    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD Calibration process in extended session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    @pytest.mark.repeat(20)
    def test_BYD_Calibration_Repeat_Phy(self,action,Driver):      
        iStep = 1
        # 写VIN
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, '2E F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17', '6E F1 90', 3, iStep,True, 'UDS', 'noClearFault')
        # 读取VIN
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 F1 90", "62 F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17", 0, iStep, False, 'UDS', 'noClearFault')
        # 清除故障并读取DTC
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "19 02 09", "59 02 09", 0, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "2E 34 01 01", "6E 34 01", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 34 01", "62 34 01 01", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "31 01 54 DF", "71 01 54 DF", 0, iStep, False, 'UDS', 'noClearFault')
        time.sleep(1.5)
        for i in range(5):
            iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "31 03 54 DF", "71 03 54 DF", 0, iStep, False, 'UDS', 'noClearFault')  
            time.sleep(1)          

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "31 02 54 DF", "71 02 54 DF", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "2E 34 01 00", "6E 34 01", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 34 01", "62 34 01 00", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "11 01", "51 01", 0, iStep, False, 'UDS', 'noClearFault')
        time.sleep(4)
        # 清除故障并读取DTC
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "19 02 09", "59 02 09", 3, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)


    @allure.severity('normal')
    @allure.story('Test read dtc information service 0x19')
    @allure.description('Test reportDTCByStatusMask of 0x19 in default session with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/62a69d6cb35db4f20e37703a/cases')
    @pytest.mark.repeat(20)
    def test_BYD_Restart_with_PowerOff_Phy(self, action, BCMPower_Gear, Driver):
        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "19 02 09", "59 02 09", 1, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)
        time.sleep(1)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "19 02 09", "59 02 09", 0, iStep, False, 'UDS', 'noClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)

    @allure.severity('normal')
    @allure.story('Test read dtc information service 0x19')
    @allure.description('Test reportDTCByStatusMask of 0x19 in default session with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/62a69d6cb35db4f20e37703a/cases')
    @pytest.mark.repeat(20)
    def test_BYD_Restart_with_1101_Phy(self, action, BCMPower_Gear, Driver):
        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "19 02 09", "59 02 09", 1, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        iStep = Driver['CAN'].restartMRR_11(Driver,iStep)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "19 02 09", "59 02 09", 0, iStep, False, 'UDS', 'noClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)

    
    @allure.severity('normal')
    @allure.story('Test read dtc information service 0x19')
    @allure.description('Test reportDTCByStatusMask of 0x19 in default session with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/62a69d6cb35db4f20e37703a/cases')
    @pytest.mark.repeat(20)
    def test_BYD_Session_switching(self, action, Driver):
        iStep = 1
        time.sleep(1)
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "10 01", "50 01", 0, iStep, False)
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "10 02", "50 02", 3, iStep, True)
        time.sleep(0.5)
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 F1 86", "62 F1 86 02", 0, iStep, False)
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "10 01", "50 01", 0, iStep, False)
        time.sleep(0.8)
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "10 03", "50 03", 0, iStep, False)
        time.sleep(0.5)
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 F1 86", "62 F1 86 03", 0, iStep, False)
        time.sleep(1)
        # 写VIN
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, '2E F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17', '6E F1 90', 0, iStep,True, 'UDS', 'noClearFault')
        # 读取VIN
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 F1 90", "62 F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17", 0, iStep, False, 'UDS', 'noClearFault')
        # 
        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "2E 34 01 00", "6E 34 01", 0, iStep, False, 'UDS', 'noClearFault')

        iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 34 01", "62 34 01 00", 0, iStep, False, 'UDS', 'noClearFault')


    @allure.severity('normal')
    @allure.story('Write data by identifier test')
    @allure.title('Test 2100 of Write data by identifier 0x2E')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    @pytest.mark.repeat(20)
    def test_WriteDataById_Extended_withPowerOff_Phy(self, action, Driver):
        iStep = 1

        Driver['CAN'].checkReqResp(Driver, "2E F1 A1 FF FF FF FF FF FF FF FF", '6E F1 A1', 3, iStep, True)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '2E 21 00 ' + Set_up.VariantCoding[0], '6E 21 00', 0, iStep, False)

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '22 21 00', '62 21 00 ' + Set_up.VariantCoding[0], 0, iStep, False)
    
        # 掉电重启后读取写入数据是否正常
        iStep = Driver['CAN'].restartMRR_Power(Driver, iStep)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '22 21 00', '62 21 00 ' + Set_up.VariantCoding[0], 1, iStep, False)

        Driver['CAN'].checkReqResp(Driver, "2E F1 A1 FF FF FF FF FF FF FF FF", '6E F1 A1', 3, iStep, True)


    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD Calibration process in extended session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_BYD_MonitorPCAN_Phy(self,action,Driver):      
        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "3E00", "7E00", 3, iStep, False, 'UDS', 'noClearFault')
        time.sleep(2)
        Driver['CAN'].checkReqResp(Driver, "2E F1 A1 FF FF FF FF FF FF FF FF", '6E F1 A1', 0, iStep, True)
        for i in range(100):
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "3E00", "7E00", 0, iStep, False, 'UDS', 'noClearFault')
            time.sleep(2)
            Driver['CAN'].checkReqResp(Driver, "2E F1 A1 FF FF FF FF FF FF FF FF", '6E F1 A1', 0, iStep, False)
            time.sleep(2)
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '2E 21 00 ' + Set_up.VariantCoding[0], '6E 21 00', 0, iStep, False)
            time.sleep(2)
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '22 21 00', '62 21 00 ' + Set_up.VariantCoding[0], 0, iStep, False)
            time.sleep(1)
            # 写VIN
            iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, '2E F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17', '6E F1 90', 0, iStep,False, 'UDS', 'noClearFault')
            time.sleep(1)
            # 读取VIN
            iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 F1 90", "62 F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17", 0, iStep, False, 'UDS', 'noClearFault')
            time.sleep(1) 
            iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "2E 34 01 00", "6E 34 01", 0, iStep, False, 'UDS', 'noClearFault')
            time.sleep(1)    
            iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 34 01", "62 34 01 00", 0, iStep, False, 'UDS', 'noClearFault')
            if project_CAN.run_faule_count != 0:
                logging.error(project_CAN.run_faule_count)
                logging.error('PCAN data error---FailedDATA:[{0}]'.format(project_CAN.corrRespStr))

        with allure.step(f'Step{iStep}: check run_faule_count == 0 # '):
                if project_CAN.run_faule_count != 0:
                    logging.error(project_CAN.run_faule_count)
                    logging.error('PCAN data error---FailedDATA:[{0}]'.format(project_CAN.corrRespStr))
                with assume: assert project_CAN.run_faule_count == 0
                iStep = iStep + 1


    ids = ["Request {}".format(binascii.b2a_hex(bytearray(req), ' '))for req, resp in SuppressPosMsg_extended]
    @allure.severity('normal')
    @allure.title('Test {req} of SuppressPosMsg in default session with Functional addressing')
    @allure.description('Test positive response of 0x10 when suppressPosMsgBit is enabled in default session with functional addressing')
    @pytest.mark.parametrize('req,resp', SuppressPosMsg_extended, ids=ids)
    def test_SuppressPosMsg_Extended_phy(self, action, Driver,req,resp):
        iStep = 1
        Driver['CAN'].checkReqResp(Driver, req, resp, 3, iStep, True)


    ids = ["Request {}".format(binascii.b2a_hex(bytearray(req), ' '))for req, resp in SuppressPosMsg_programming]
    @allure.severity('normal')
    @allure.title('Test {req} of SuppressPosMsg in default session with Functional addressing')
    @allure.description('Test positive response of 0x10 when suppressPosMsgBit is enabled in default session with functional addressing')
    @pytest.mark.parametrize('req,resp', SuppressPosMsg_programming, ids=ids)
    def test_SuppressPosMsg_Programming_phy(self, action, Driver,req,resp):
        iStep = 1
        Driver['CAN'].checkReqResp(Driver, req, resp, 2, iStep, True)


    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD Shouhou Calibration process in extended session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_BYD_Shouhou_Calibration_LowSpeed_Phy(self,action,Driver):      
        iStep = 1
        # 开启SDA校准
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '31 01 03 02', '71 01 03 02', 3, iStep,True, 'UDS', 'noClearFault')
        
        # 读取校准进度和结果
        for i in range(2):
            #设置车速大于3KM/H
            iStep = Driver['CAN'].setSignalValue('IPB_0x121','IPB_Vehicle_Speed_S',2,iStep)
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "31 03 03 02", "71 03 03 02", 0, iStep, False, 'UDS', 'noClearFault')
            time.sleep(1)

        with assume: assert uds_resp[5] & 0x01 == 0x01
        # 中止SDA校准
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "31 02 03 02", "71 02 03 02", 0, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].setSignalValue('IPB_0x121','IPB_Vehicle_Speed_S',0,iStep)


    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD Shouhou Calibration process in extended session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_BYD_Shouhou_Calibration_OverSpeed_Phy(self,action,Driver):      
        iStep = 1
        # 开启SDA校准
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '31 01 03 02', '71 01 03 02', 3, iStep,True, 'UDS', 'noClearFault')
        
        # 读取校准进度和结果
        for i in range(5):
            #设置车速大于3KM/H
            iStep = Driver['CAN'].setSignalValue('IPB_0x121','IPB_Vehicle_Speed_S',280,iStep)
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "31 03 03 02", "71 03 03 02", 0, iStep, False, 'UDS', 'noClearFault')
            if uds_resp[5] & 0x02 == 0x02:
                break
            time.sleep(1)

        with assume: assert uds_resp[5] & 0x02 == 0x02 
        # 中止SDA校准
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "31 02 03 02", "71 02 03 02", 0, iStep, False, 'UDS', 'ClearFault')

        iStep = Driver['CAN'].setSignalValue('IPB_0x121','IPB_Vehicle_Speed_S',0,iStep)


    @allure.severity('normal')
    @allure.title('Test Read_MRR_Model_From64E in default session with physical addressing')
    @allure.description('Test Read_MRR_Model_From64E in default session with physical addressing')
    @pytest.mark.repeat(200)
    def test_Read_MRR_Model_From64E_phy(self, action, Driver):
        iStep = 1
        Driver['CAN'].checkReqResp(Driver, '10 01', '50 01', 0, iStep)

        Driver['CAN2'].fush_q_buff()  # 清缓存  
        time.sleep(0.1)
        ALl_data,buffer_size = Driver['CAN2'].read_frame()
        response_data = ALl_data[ALl_data['AIdentifier'] == 0x64E]
        respData = response_data.iloc[0]['ADatas'][:16]

        actuRespStr = binascii.b2a_hex(bytearray(respData), ' ') 
        logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x64E, actuRespStr)) 
        MRR_Model = respData[3] & 0x0F
        logging.info('MRR Model:[{0}]'.format(MRR_Model)) 

        with assume: assert MRR_Model == 3
		
	@allure.severity('normal')
    @allure.title('Test Read_MRR_HostSpeedX_From64E in default session with physical addressing')
    @allure.description('Test Read_MRR_HostSpeedX_From64E in default session with physical addressing')
    @pytest.mark.repeat(200)
    def test_Read_MRR_HostSpeedX_From64E_phy(self, action, Driver):
        iStep = 1
        Driver['CAN'].checkReqResp(Driver, '10 01', '50 01', 0, iStep)
        #设置车速大于3KM/H
        iStep = Driver['CAN'].setSignalValue('IPB_0x121','IPB_Vehicle_Speed_S',20,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x1F0','Wheel_Speed_FL_1F0_S',20,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x1F0','Wheel_Speed_FR_1F0_S',20,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x1F0','Wheel_Speed_RL_1F0_S',20,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x1F0','Wheel_Speed_RR_1F0_S',20,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x223','IPB_AX_offset_S',0,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x223','IPB_AX_S',0,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x223','IPB_AY_Offset_S',0,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x223','IPB_AY_S',0,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x222','Yaw_Rate_Offset_S',0,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x222','Yaw_Rate_Signal_S',0,iStep)
        # iStep = Driver['CAN'].setSignalValue('IPB_0x222','Yaw_Rate_Status_S',0,iStep)

        Driver['CAN2'].fush_q_buff()  # 清缓存  
        time.sleep(0.1)
        ALl_data,buffer_size = Driver['CAN2'].read_frame()
        response_data = ALl_data[ALl_data['AIdentifier'] == 0x64E]
        respData = response_data.iloc[0]['ADatas'][:16]

        actuRespStr = binascii.b2a_hex(bytearray(respData), ' ') 
        logging.info('P_CAN responseID:[0x{0:X}], Data:[{1}]'.format(0x64E, actuRespStr))
        byte2 = respData[6] & 0x10
        byte1 = respData[5]
        packed = byte2.to_bytes(1, 'little') + byte1.to_bytes(1, 'little')
        HostSpeedX = int.from_bytes(packed, byteorder = 'big') + 0.0
        logging.info('HostSpeedX:[{0}]'.format(HostSpeedX))
        HostSpeedX = (HostSpeedX * 0.03125 -128) * 3600 / 1000  
        logging.info('MRR HostSpeedX:[{0}]'.format(HostSpeedX)) 

        with assume: assert 19 < HostSpeedX < 21

        iStep = Driver['CAN'].setSignalValue('IPB_0x121','IPB_Vehicle_Speed_S',0,iStep)
        
    @allure.severity('normal')
    @allure.title('Test Read_MRR_signal at voltage is 18v in default session with physical addressing')
    @allure.description('Test Read_MRR_signal at voltage is 18v in default session with physical addressing')
    @pytest.mark.repeat(10)
    def test_Read_MRR_Signal_18V_phy(self, action, Driver):
        iStep = 1

        iStep = Driver['CAN'].SetVolt(Driver, 18, 0.5, iStep)
        Driver['CAN'].checkReqResp(Driver, '10 01', '50 01', 0, iStep)

        self.checkPcanSignal(Driver)

        iStep = Driver['CAN'].SetVolt(Driver, 12, 0.5, iStep)

    
    @allure.severity('normal')
    @allure.title('Test Read_MRR_signal at voltage is 6v in default session with physical addressing')
    @allure.description('Test Read_MRR_signal at voltage is 6v in default session with physical addressing')
    @pytest.mark.repeat(10)
    def test_Read_MRR_Signal_8V_phy(self, action, Driver):
        iStep = 1

        iStep = Driver['CAN'].SetVolt(Driver, 8, 0.5, iStep)
        Driver['CAN'].checkReqResp(Driver, '10 01', '50 01', 0, iStep)

        self.checkPcanSignal(Driver)

        iStep = Driver['CAN'].SetVolt(Driver, 12, 0.5, iStep)



    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD Calibration process in default session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_BYD_MonitorPCAN_EDR_Busoff_Phy(self,action,Driver):      
        iStep = 1
        for j in range(2):
            for i in range(30):
                iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "3E00", "7E00", 3, iStep, False, 'UDS', 'noClearFault')
                Driver['CAN'].checkReqResp(Driver, "2E F1 A1 FF FF FF FF FF FF FF FF", '6E F1 A1', 0, iStep, True)
                iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, "3E00", "7E00", 0, iStep, False, 'UDS', 'noClearFault')
                time.sleep(2)
                Driver['CAN'].checkReqResp(Driver, "2E F1 A1 FF FF FF FF FF FF FF FF", '6E F1 A1', 0, iStep, False)
                time.sleep(2)
                iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '2E 21 00 ' + Set_up.VariantCoding[0], '6E 21 00', 0, iStep, False)
                time.sleep(2)
                iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '22 21 00', '62 21 00 ' + Set_up.VariantCoding[0], 0, iStep, False)
                time.sleep(1)
                # 写VIN
                iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, '2E F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17', '6E F1 90', 0, iStep,False, 'UDS', 'noClearFault')
                time.sleep(1)
                # 读取VIN
                iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 F1 90", "62 F1 90 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17", 0, iStep, False, 'UDS', 'noClearFault')
                time.sleep(1) 
                iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "2E 34 01 00", "6E 34 01", 0, iStep, False, 'UDS', 'noClearFault')
                time.sleep(1)    
                iStep,uds_resp, size = Driver['CAN'].checkReqRespNoWait(Driver, "22 34 01", "62 34 01 00", 0, iStep, False, 'UDS', 'noClearFault')
                if project_CAN.run_faule_count != 0:
                    logging.error(project_CAN.run_faule_count)
                    logging.error('PCAN data error---FailedDATA:[{0}]'.format(project_CAN.corrRespStr))

                with allure.step(f'Step{iStep}: check run_faule_count == 0 # '):
                    if project_CAN.run_faule_count != 0:
                        logging.error(project_CAN.run_faule_count)
                        logging.error('PCAN data error---FailedDATA:[{0}]'.format(project_CAN.corrRespStr))
                    with assume: assert project_CAN.run_faule_count == 0
                    iStep = iStep + 1

            Driver['CAN'].S_ConnectCanH_CanL(True, iStep)
            Driver['CAN'].S_ConnectCanH_CanL(False, iStep)
            iStep = self.searchDTC(Driver, [0x19,0x02,0x09], [0xC1,0xD7,0x88,0x08], iStep, True, 500, 1)

            Driver['CAN'].V_ConnectCanH_CanL(True, iStep)
            Driver['CAN'].V_ConnectCanH_CanL(False, iStep)

            iStep = self.searchDTC(Driver, [0x19,0x02,0x09], [0xC1,0xD7,0x88,0x08], iStep, True, 500, 1)

            Driver['CAN'].checkReqResp(Driver, "14 FF FF FF", '54', 0, iStep, False)


    ids = ["Check 0x{:X} cycle".format(ID)for ID in Set_up.PCAN_list_BOCH.keys()]
    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD pcan BOCH data cycle in default session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    # @pytest.mark.repeat(10)
    @pytest.mark.parametrize('ID', Set_up.PCAN_list_BOCH.keys(), ids=ids)
    def test_Pcan_Data_Cycle_BOCH_phy(self,action,Driver, ID):
        for i in range(20):
            fps = Driver['CAN2'].get_fps(ID, 'CANFD')
            logging.info('ID (0x{0:X}) fps---{1}'.format(ID, fps))
            with assume: assert 17 < fps < 22
            time.sleep(1)


    @allure.severity('normal')
    @allure.story('BYD Calibration test')
    @allure.description('Test BYD pcan BOCH data interval no EDR in default session with physical addressing')
    @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    @pytest.mark.repeat(10)
    def test_Pcan_Data_interval_BOCH_phy(self,action,Driver):
        self.checkSignalInterval(Driver, 5, True, Set_up.PCAN_list_BOCH)

    # @allure.severity('normal')
    # @allure.story('BYD Calibration test')
    # @allure.description('Test BYD pcan BOCH data interval no EDR in default session with physical addressing')
    # @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    # @pytest.mark.repeat(10)
    # def test_Pcan_Data_interval_BOCH_noEDR_phy(self,action,Driver):
    #     self.checkSignalInterval(Driver, 5, False, Set_up.PCAN_list_BOCH)

    
    # ids = ["Check 0x{:X} cycle".format(ID)for ID in Set_up.PCAN_list.keys()]
    # @allure.severity('normal')
    # @allure.story('BYD Calibration test')
    # @allure.description('Test BYD pcan dalu data cycle in default session with physical addressing')
    # @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    # @pytest.mark.parametrize('ID', Set_up.PCAN_list.keys(), ids=ids)
    # def test_Pcan_Data_Cycle_dalu_phy(self,action,Driver, ID):
    #     for i in range(20):
    #         fps = Driver['CAN2'].get_fps(ID, 'CANFD')
    #         logging.info('ID (0x{0:X}) fps---{1}'.format(ID, fps))
    #         with assume: assert 18 < fps < 21
    #         time.sleep(1)


    # @allure.severity('normal')
    # @allure.story('BYD Calibration test')
    # @allure.description('Test BYD pcan dalu data interval in default session with physical addressing')
    # @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    # @pytest.mark.repeat(10)
    # def test_Pcan_Data_interval_dalu_noEDR_phy(self,action,Driver):
    #     self.checkSignalInterval(Driver, 5, False, Set_up.PCAN_list)

    # @allure.severity('normal')
    # @allure.story('BYD Calibration test')
    # @allure.description('Test BYD pcan dalu data interval in default session with physical addressing')
    # @allure.testcase('https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    # @pytest.mark.repeat(10)
    # def test_Pcan_Data_interval_dalu_phy(self,action,Driver):
    #     self.checkSignalInterval(Driver, 5, True, Set_up.PCAN_list)

    def checkSignalInterval(self,Driver, timeLenth = 5, edrType = False, dictSouce = Set_up.PCAN_list_BOCH):
        failCount = 0

        Driver['CAN2'].fush_q_buff()  # 清缓存
        time.sleep(timeLenth)
        ALl_data,buffer_size = Driver['CAN2'].read_frame()
        for ID in dictSouce:
            response_data = ALl_data[ALl_data['AIdentifier']== ID]
            count = len(response_data)
            timeInterval = 0
            for i in range (count -1):
                T1 = round(response_data.iloc[i]['ATimeUs'] * 1000, 3)
                T2 = round(response_data.iloc[i+1]['ATimeUs'] * 1000, 3)
                timeInterval = round((T2-T1), 3)
                    
                if edrType ==  False:
                    if 45 > timeInterval and  timeInterval > 55:
                        logging.error('第[{2:02d}]组, ID = [0x{0:X}], time_interval = [{1:.3f}]ms, T2 = [{3:.3f}], T1 = [{4:.3f}]'.format(ID,timeInterval, i, T2, T1))
                        failCount += 1
                    else:
                        logging.info('第[{2:02d}]组, ID = [0x{0:X}], time_interval = [{1:.3f}]ms, T2 = [{3:.3f}], T1 = [{4:.3f}]'.format(ID,timeInterval, i, T2, T1))
                else:
                    if 40 > timeInterval and  timeInterval > 60:
                        logging.error('第[{2:02d}]组, ID = [0x{0:X}], time_interval = [{1:.3f}]ms, T2 = [{3:.3f}], T1 = [{4:.3f}]'.format(ID,timeInterval, i, T2, T1))
                        failCount += 1
                    else:
                        logging.info('第[{2:02d}]组, ID = [0x{0:X}], time_interval = [{1:3f}]ms, T2 = [{3:3f}], T1 = [{4:3f}]'.format(ID,timeInterval, i, T2, T1))

        logging.info('帧间隔超出范围次数[{0}]'.format(failCount))
        
        with assume: assert failCount < 5



    def checkPcanSignal(self,Driver):
        Driver['CAN2'].fush_q_buff()  # 清缓存
        time.sleep(1)
        ALl_data,buffer_size = Driver['CAN2'].read_frame()
        response_data_80 = ALl_data[ALl_data['AIdentifier']== 0x80]
        response_data_66F = ALl_data[ALl_data['AIdentifier']== 0x66F]

        count_80 = len(response_data_80)
        receive_data = []
        try:
            last_data = response_data_66F.iloc[0]['ADatas'] 
            for i in range (8):
                datai = last_data[i]
                receive_data.append(datai)
        except:
            last_data = []
        readstr = binascii.b2a_hex(bytearray(receive_data),' ')
        logging.info('receive_ID = [{0:X}], receive_Data = [{1}], count = [{2}]'.format(0x80,readstr, count_80))

        count_66F = len(response_data_66F)
        receive_data = []
        try:
            last_data = response_data_66F.iloc[0]['ADatas'] 
            for i in range (8):
                datai = last_data[i]
                receive_data.append(datai)
        except:
            last_data = []
        readstr = binascii.b2a_hex(bytearray(receive_data),' ')
        logging.info('receive_ID = [{0:X}], receive_Data = [{1}], count = [{2}]'.format(0x66F,readstr, count_66F))
        with assume: assert (count_66F > 0 or count_80 > 0)

    
    def searchDTC(self, Driver,req, DTC_Number,iStep, isTrue, MatureConditionsTime=1000, timeout=0.5):
        # 查询故障
        with allure.step(f"Step{iStep}: search DTC {binascii.b2a_hex(bytearray(DTC_Number), ' ')}"):
            time.sleep(MatureConditionsTime/1000)
            a,b,c=Driver['UDS'].search_DTC(req, DTC_Number, timeout)
            if isTrue:
                if (b + MatureConditionsTime/1000) > (MatureConditionsTime/1000 + 1):
                    logging.error(f"MatureConditionsTime = {MatureConditionsTime/1000 + 1}, realtime = {b + MatureConditionsTime/1000}")
                if a != True:
                    logging.error(f"No found DTC{binascii.b2a_hex(bytearray(DTC_Number), ' ')}")
                with assume: assert a == True and (b + MatureConditionsTime/1000) <= (MatureConditionsTime/1000 + 1)
            else:
                with assume: assert a == False
            iStep +=1
        return iStep